package com.example.sendmessage.server.dao;

import com.example.sendmessage.common.MqException;
import com.example.sendmessage.server.core.Binding;
import com.example.sendmessage.server.core.Exchange;
import com.example.sendmessage.server.core.Message;
import com.example.sendmessage.server.core.Queue;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 数据在内存中的管理
 */
public class MemoryDataCenter {
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    private ConcurrentHashMap<String, Queue> queueMap = new ConcurrentHashMap<>();
    // 第一个String是交换机的名字，第二个String是队列的名字
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
    // 第一个String是队列的名字，第二个String是待确认消息的id
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> queueMessageWaitAck = new ConcurrentHashMap<>();

    /**
     * 内存中添加一个交换机
     * @param exchange 添加的交换机
     */
    public void insertExchange(Exchange exchange) {
        exchangeMap.put(exchange.getName(), exchange);
        System.out.println("[MemoryDataCenter] 交换机添加成功！exchangeName = " + exchange.getName());
    }

    /**
     * 删除交换机
     * @param exchangeName 删除的交换机的名字
     */
    public void deleteExchange(String exchangeName) {
        exchangeMap.remove(exchangeName);
        System.out.println("[MemoryDataCenter] 交换机删除成功！exchangeName = " + exchangeName);
    }

    /**
     * 查找一个交换机
     * @param exchangeName 查找的交换机名
     * @return 查找到的交换机
     */
    public Exchange getExchange(String exchangeName) {
        return exchangeMap.get(exchangeName);
    }

    /**
     * 添加一个队列
     * @param queue 添加的队列
     */
    public void insertQueue(Queue queue) {
        queueMap.put(queue.getName(), queue);
        System.out.println("[MemoryDataCenter] 队列添加成功！queueName = " + queue.getName());
    }

    /**
     * 删除一个队列
     * @param queueName 删除的队列名
     */
    public void deleteQueue(String queueName) {
        queueMap.remove(queueName);
        System.out.println("[MemoryDataCenter] 队列删除成功！queueName = " + queueName);
    }

    /**
     * 获取一个队列
     * @param queueName 获取队列的队列名
     * @return 获取的队列
     */
    public Queue getQueue(String queueName) {
        return queueMap.get(queueName);
    }

    /**
     * 添加绑定关系
     * @param binding 添加的绑定
     * @throws MqException 绑定存在的异常
     */
    public void insertBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(
                binding.getExchangeName(), k-> new ConcurrentHashMap<>());
        synchronized (bindingMap) {
            if (bindingMap.get(binding.getQueueName()) != null) {
                // 这里也是参考rabbitMQ不能修改和重复添加绑定
                throw new MqException("[MemoryDataCenter] 绑定已经存在！exchangeName = " +
                        binding.getExchangeName() + "queueName = " + binding.getQueueName());
            }
            bindingMap.put(binding.getQueueName(), binding);
            System.out.println("[MemoryDataCenter] 绑定添加成功！exchangeName = " +
                    binding.getExchangeName() + "，queueName = " + binding.getQueueName());
        }
    }

    /**
     * 删除绑定
     * @param binding 需要删除的绑定
     * @throws MqException 没有绑定抛出异常
     */
    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if (bindingMap == null) {
            // 没有绑定，不能删除
            throw new MqException("[MemoryDataCenter] 交换机没有绑定队列！exchangeName = " + binding.getExchangeName());
        }
        bindingMap.remove(binding.getQueueName());
        System.out.println("[MemoryDataCenter] 绑定解除成功！exchangeName = " +
                binding.getExchangeName() + "，queueName = " + binding.getQueueName());
    }

    /**
     * 获取一个绑定
     * @param exchangeName 交换机名字
     * @param queueName 队列名字
     * @return 一个绑定关系
     */
    public Binding getBinding(String exchangeName, String queueName) {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if (bindingMap == null) {
            return null;
        }
        return bindingMap.get(queueName);
    }

    /**
     * 返回一个交换机的所有绑定
     * @param exchangeName 交换机名字
     * @return 所有的绑定
     */
    public ConcurrentHashMap<String, Binding> getBindings(String exchangeName) {
        return bindingsMap.get(exchangeName);
    }

    /**
     * 将消息添加到消息中心
     * @param message 即将添加的消息
     */
    public void insertMessage(Message message) {
        messageMap.put(message.getMessageId(), message);
        System.out.println("[MemoryDataCenter] 消息中心添加消息成功！MessageId = " + message.getMessageId());
    }

    /**
     * 删除消息中心的消息
     * @param messageId 删除的消息id
     */
    public void deleteMessage(String messageId) {
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息中心移除消息成功！MessageId = " + messageId);
    }

    /**
     * 根据消息id获取消息内容
     * @param messageId 消息唯一id
     * @return 查询到的消息
     */
    public Message getMessage(String messageId) {
        return messageMap.get(messageId);
    }

    /**
     * 将消息发送到指定的队列
     * @param queue 发送的队列
     * @param message 发送的消息
     */
    public void sendMessage(Queue queue, Message message) {
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queue.getName(),
                k-> new LinkedList<>());
        synchronized (queue) {
            // 先将消息存入消息中心
            messages.add(message);
            messageMap.put(message.getMessageId(), message);
        }
        System.out.println("[MemoryDataCenter] 消息发送到指定队列成功！messageId = " +
                message.getMessageId() + "，queueName = " + queue.getName());
    }

    /**
     * 获取队列中的消息个数，用于判断推送时是否完全推送
     * 或者判断建立推送关系时队列中是否含有消息
     * @param queueName 队列名
     * @return 队列中的消息个数
     */
    public int getQueueMessageNum(String queueName) {
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) {
            return 0;
        }
        synchronized (messages) {
            return messages.size();
        }
    }

    /**
     * 队列推送一个消息
     * @param queueName 推送消息的队列
     * @return 推送的消息
     */
    public Message pollMessage(String queueName) {
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null || messages.size() == 0) {
            return null;
        }
        Message message = messages.remove(0);
        System.out.println("[MemoryDataCenter] 消息消费成功！queueName = " + queueName +
                "，messageId = " + message.getMessageId());
        return message;
    }

    /**
     * 可靠性的安全机制，等待消费者反馈才能删除消息
     * @param queueName 队列名
     * @param message 待确认的消息
     */
    public void insertMessageWaitAck(String queueName, Message message) {
        ConcurrentHashMap<String, Message> messageMap = queueMessageWaitAck.computeIfAbsent(
                queueName, k -> new ConcurrentHashMap<>());
        messageMap.put(message.getMessageId(), message);
        System.out.println("[MemoryDataCenter] 消息等待确认！messageId = " + message.getMessageId());
    }

    /**
     * 确认了消息之后就删除
     * @param queueName 队列名
     * @param messageId 已确认的消息id
     */
    public void deleteMessageWaitAck(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> messageMap = queueMessageWaitAck.get(queueName);
        if (messageMap == null || messageMap.size() == 0) {
            // 消息为空就不用删除了
            return;
        }
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息确认成功！messageId = " + messageId);
    }

    /**
     * 获取指定队列中未确认的消息
     * 看是重发消息给客户端还是删除消息（待扩展，死信队列）
     * @param queueName 队列名
     * @param messageId 未确认的消息id
     * @return 未确认的消息
     */
    public Message getMessageWaitAck(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> messageMap = queueMessageWaitAck.get(queueName);
        if (messageMap == null || messageMap.size() == 0) {
            return null;
        }
        return messageMap.get(messageId);
    }

    /**
     * 内存消息的恢复机制，反正因为宕机造成数据丢失
     * @param diskDataCenter 硬盘数据中心
     */
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {
        // 防止残留数据
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        messageMap.clear();
        queueMessageMap.clear();
        queueMessageWaitAck.clear();
        // 从硬盘恢复交换机的数据
        List<Exchange> exchangeList = diskDataCenter.selectAllExchanges();
        for (Exchange exchange : exchangeList) {
            exchangeMap.put(exchange.getName(), exchange);
        }
        // 从硬盘恢复队列数据
        List<Queue> queueList = diskDataCenter.selectAllQueues();
        for (Queue queue : queueList) {
            queueMap.put(queue.getName(), queue);
        }
        // 恢复绑定关系
        List<Binding> bindingList = diskDataCenter.selectAllBindings();
        for (Binding binding : bindingList) {
            ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(
                    binding.getExchangeName(), k -> new ConcurrentHashMap<>());
            bindingMap.put(binding.getQueueName(), binding);
        }
        // 恢复所有的消息数据，每个队列有每个队列的消息
        for (Queue queue : queueList) {
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue.getName());
            queueMessageMap.put(queue.getName(), messages);
            // 消息中心的数据也要恢复
            for (Message message : messages) {
                messageMap.put(message.getMessageId(), message);
            }
        }
        System.out.println("[MemoryDataCenter] 内存消息恢复成功！");
    }
}
